// 版权所有2009 Go作者。版权所有。
// 此源代码的使用受BSD样式
// 许可证的约束，该许可证可以在许可证文件中找到。

package rpc

import (
	"bufio"
	"encoding/gob"
	"errors"
	"io"
	"log"
	"net"
	"net/http"
	"sync"
)

// ServerError表示从RPC连接的远程端返回的错误。
type ServerError string

func (e ServerError) Error() string {
	return string(e)
}

var ErrShutdown = errors.New("connection is shut down")

// 调用表示活动RPC。
type Call struct {
	ServiceMethod string      // 要调用的服务和方法的名称。
	Args          interface{} // 函数（*struct）的参数。
	Reply         interface{} // 函数（*struct）的回复。
	Error         error       // 完成后，返回错误状态。
	Done          chan *Call  // Go完成时接收*呼叫。
}

// 客户端表示RPC客户端。
// 可能有多个未完成的呼叫与单个客户端关联，并且一个客户端可能被
// 多个goroutine同时使用。
type Client struct {
	codec ClientCodec

	reqMutex sync.Mutex // 保护后续
	request  Request

	mutex    sync.Mutex // 保护后续
	seq      uint64
	pending  map[uint64]*Call
	closing  bool // 用户已呼叫Close 
	shutdown bool // 服务器已通知我们停止
}

// 客户端编解码器实现RPC请求的写入和RPC会话客户端RPC响应的读取。
// 客户端调用WriteRequest向连接写入请求
// 并成对调用ReadResponseHeader和ReadResponseBody 
// 读取响应。完成
// 连接后，客户端调用Close。ReadResponseBody可以使用nil 
// 参数调用，以强制读取响应体，然后放弃
// 。
// 有关并发访问的信息，请参阅NewClient的注释。
type ClientCodec interface {
	WriteRequest(*Request, interface{}) error
	ReadResponseHeader(*Response) error
	ReadResponseBody(interface{}) error

	Close() error
}

func (client *Client) send(call *Call) {
	client.reqMutex.Lock()
	defer client.reqMutex.Unlock()

	// 注册此呼叫。
	client.mutex.Lock()
	if client.shutdown || client.closing {
		client.mutex.Unlock()
		call.Error = ErrShutdown
		call.done()
		return
	}
	seq := client.seq
	client.seq++
	client.pending[seq] = call
	client.mutex.Unlock()

	// 编码并发送请求。
	client.request.Seq = seq
	client.request.ServiceMethod = call.ServiceMethod
	err := client.codec.WriteRequest(&client.request, call.Args)
	if err != nil {
		client.mutex.Lock()
		call = client.pending[seq]
		delete(client.pending, seq)
		client.mutex.Unlock()
		if call != nil {
			call.Error = err
			call.done()
		}
	}
}

func (client *Client) input() {
	var err error
	var response Response
	for err == nil {
		response = Response{}
		err = client.codec.ReadResponseHeader(&response)
		if err != nil {
			break
		}
		seq := response.Seq
		client.mutex.Lock()
		call := client.pending[seq]
		delete(client.pending, seq)
		client.mutex.Unlock()

		switch {
		case call == nil:
			// 我们没有挂起的电话。这通常意味着
			// WriteRequest部分失败，调用已被
			// 删除；响应是一个服务器，它告诉我们读取请求正文时发生
			// 错误。我们仍然应该尝试
			// 来读取错误体，但是没有人可以给出它。
			err = client.codec.ReadResponseBody(nil)
			if err != nil {
				err = errors.New("reading error body: " + err.Error())
			}
		case response.Error != "":
			// 我们收到一个错误响应。把这个交给请求人；
			// 任何后续请求都将获得ReadResponseBy 
			// 错误（如果有）。
			call.Error = ServerError(response.Error)
			err = client.codec.ReadResponseBody(nil)
			if err != nil {
				err = errors.New("reading error body: " + err.Error())
			}
			call.done()
		default:
			err = client.codec.ReadResponseBody(call.Reply)
			if err != nil {
				call.Error = errors.New("reading body " + err.Error())
			}
			call.done()
		}
	}
	// 终止挂起的调用。
	client.reqMutex.Lock()
	client.mutex.Lock()
	client.shutdown = true
	closing := client.closing
	if err == io.EOF {
		if closing {
			err = ErrShutdown
		} else {
			err = io.ErrUnexpectedEOF
		}
	}
	for _, call := range client.pending {
		call.Error = err
		call.done()
	}
	client.mutex.Unlock()
	client.reqMutex.Unlock()
	if debugLog && err != io.EOF && !closing {
		log.Println("rpc: client protocol error:", err)
	}
}

func (call *Call) done() {
	select {
	case call.Done <- call:
		// 好的
	default:
		// 我们不想在这里阻塞。调用者有责任确保
		// 通道有足够的缓冲空间。请参阅Go（）中的注释。
		if debugLog {
			log.Println("rpc: discarding Call reply due to insufficient Done chan capacity")
		}
	}
}

// NewClient返回一个新的客户端来处理对连接另一端的一组服务的请求。
// 它在连接的写端添加了一个缓冲区，因此
// 头和有效负载作为一个单元发送。
// 
// 连接的读写部分是独立序列化的，
// 因此不需要联锁。但是，每一半都可以并发访问
// 因此conn的实现应该能够防止
// 并发读取或并发写入。
func NewClient(conn io.ReadWriteCloser) *Client {
	encBuf := bufio.NewWriter(conn)
	client := &gobClientCodec{conn, gob.NewDecoder(conn), gob.NewEncoder(encBuf), encBuf}
	return NewClientWithCodec(client)
}

// NewClientWithCodec与NewClient类似，但使用指定的
// codec对请求进行编码并对响应进行解码。
func NewClientWithCodec(codec ClientCodec) *Client {
	client := &Client{
		codec:   codec,
		pending: make(map[uint64]*Call),
	}
	go client.input()
	return client
}

type gobClientCodec struct {
	rwc    io.ReadWriteCloser
	dec    *gob.Decoder
	enc    *gob.Encoder
	encBuf *bufio.Writer
}

func (c *gobClientCodec) WriteRequest(r *Request, body interface{}) (err error) {
	if err = c.enc.Encode(r); err != nil {
		return
	}
	if err = c.enc.Encode(body); err != nil {
		return
	}
	return c.encBuf.Flush()
}

func (c *gobClientCodec) ReadResponseHeader(r *Response) error {
	return c.dec.Decode(r)
}

func (c *gobClientCodec) ReadResponseBody(body interface{}) error {
	return c.dec.Decode(body)
}

func (c *gobClientCodec) Close() error {
	return c.rwc.Close()
}

// 拨号HTTP连接到指定网络地址的HTTP RPC服务器
// 在默认HTTP RPC路径上侦听。
func DialHTTP(network, address string) (*Client, error) {
	return DialHTTPPath(network, address, DefaultRPCPath)
}

// DialHTTPPath在指定的网络地址和路径连接到HTTP RPC服务器
// 上。
func DialHTTPPath(network, address, path string) (*Client, error) {
	conn, err := net.Dial(network, address)
	if err != nil {
		return nil, err
	}
	io.WriteString(conn, "CONNECT "+path+" HTTP/1.0\n\n")

	// 需要成功的HTTP响应
	// 才能切换到RPC协议。
	resp, err := http.ReadResponse(bufio.NewReader(conn), &http.Request{Method: "CONNECT"})
	if err == nil && resp.Status == connected {
		return NewClient(conn), nil
	}
	if err == nil {
		err = errors.New("unexpected HTTP response: " + resp.Status)
	}
	conn.Close()
	return nil, &net.OpError{
		Op:   "dial-http",
		Net:  network + " " + address,
		Addr: nil,
		Err:  err,
	}
}

// 拨号连接到指定网络地址的RPC服务器。
func Dial(network, address string) (*Client, error) {
	conn, err := net.Dial(network, address)
	if err != nil {
		return nil, err
	}
	return NewClient(conn), nil
}

// Close调用底层编解码器的Close方法。如果连接已经
// 正在关闭，则返回ErrShutdown。
func (client *Client) Close() error {
	client.mutex.Lock()
	if client.closing {
		client.mutex.Unlock()
		return ErrShutdown
	}
	client.closing = true
	client.mutex.Unlock()
	return client.codec.Close()
}

// Go异步调用函数。它返回表示调用的
// 的调用结构。当调用完成时，done通道将通过返回
// 同一调用对象发出信号。如果done为nil，Go将分配一个新通道。
// 如果非nil，则必须缓冲done，否则Go将故意崩溃。
func (client *Client) Go(serviceMethod string, args interface{}, reply interface{}, done chan *Call) *Call {
	call := new(Call)
	call.ServiceMethod = serviceMethod
	call.Args = args
	call.Reply = reply
	if done == nil {
		done = make(chan *Call, 10) // 已缓冲。
	} else {
		// 如果调用方通过，则完成！=无，必须安排
		// done有足够的缓冲区，用于同时使用该通道的
		// RPC的数量。如果频道
		// 完全没有缓冲，最好不要运行。
		if cap(done) == 0 {
			log.Panic("rpc: done channel is unbuffered")
		}
	}
	call.Done = done
	client.send(call)
	return call
}

// 调用调用命名函数，等待其完成，并返回其错误状态。
func (client *Client) Call(serviceMethod string, args interface{}, reply interface{}) error {
	call := <-client.Go(serviceMethod, args, reply, make(chan *Call, 1)).Done
	return call.Error
}
